package com.practise.rocketmqdemo.controller;

import com.alibaba.fastjson.JSON;
import com.practise.rocketmqdemo.constant.MQConstant;
import com.practise.rocketmqdemo.entity.User;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Arrays;
import java.util.List;
import java.util.UUID;

@Slf4j
@RestController
@RequestMapping(value = "topic")
@Api(value = "send topic Msg controller", tags = {"topic消息队列接口"})
public class TopicMsgController {

    @Autowired
    RocketMQTemplate mqTemplate;

    @ApiOperation(value = "发送消息到topic Q1", httpMethod = "POST")
    @RequestMapping(value = "sendToQueue1")
    public void sendToQueue1(@RequestBody @ApiParam(name = "User", value = "用户信息", required = true) User user, int msgType) {

        String msg = JSON.toJSONString(user);
        Message<String> message = MessageBuilder.withPayload(msg).build();
        try {
            switch (msgType) {
                case 1:
                    //普通发送1
                    mqTemplate.convertAndSend(MQConstant.TOPIC_ONE, msg);
                    break;
                case 2:
                    //普通发送2
                    mqTemplate.send(MQConstant.TOPIC_ONE, message);
                    break;
                case 3:
                    //单向发送3 只发送请求不等待应答,适用于耗时非常短，但对可靠性要求并不高的场景，例如日志收集
                    mqTemplate.sendOneWay(MQConstant.TOPIC_ONE, msg);
                    break;
                case 4:
                    //批量消息
                    List<Message<String>> messageList = Arrays.asList(
                            MessageBuilder.withPayload("aaaa").build(), MessageBuilder.withPayload("bbbb").build());
                    mqTemplate.syncSend(MQConstant.TOPIC_ONE,messageList);
                    break;
                case 5:
                    /**
                     ** 同步延迟消息
                     ** rocketMQ的延迟消息发送其实是已发送就已经到broker端了，然后消费端会延迟收到消息。
                     ** RocketMQ 目前只支持固定精度的定时消息。
                     ** 固定等级：1到18分别对应1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
                     ** 延迟的底层方法是用定时任务实现的。
                     ** 发送延时消息（delayLevel的值就为0，因为不延时）
                     * **/
                    mqTemplate.syncSend(MQConstant.TOPIC_ONE, message,5000,3);
                    break;
                case 6:
                    //异步延迟消息
                    mqTemplate.asyncSend(MQConstant.TOPIC_ONE, message, new SendCallback() {
                        @Override
                        public void onSuccess(SendResult sendResult) {
                            log.info("send MQConstant.TOPIC_ONE success {} ",sendResult);
                        }

                        @Override
                        public void onException(Throwable e) {
                            log.error("send MQConstant.TOPIC_ONE onException ",e);
                        }
                    },5000,3);
                    break;
                case 7:
                    // 支持同步\异步\单向发送顺序消息
                    //hashKey参数，主要用来计算选择队列的，一般可以把订单ID，产品ID作为参数值,制定顺序规则送进队列；
                    // 消费端接收的时候，默认是并发多线程去接收消息,RocketMQMessageListener有个属性consumeMode，默认是ConsumeMode.CONCURRENTLY,
                    // 需改成ConsumeMode.ORDERLY，单线程顺序接收消息
                    String haskKey;
                    for (int i = 0; i < 10; i++) {
                        haskKey= String.valueOf(System.currentTimeMillis());
                        message = MessageBuilder.withPayload(msg.concat("_"+i)).build();
                        mqTemplate.syncSendOrderly(MQConstant.TOPIC_ORDER, message,haskKey);
                        Thread.sleep(1000);
                    }
                    break;
            }
        }catch (InterruptedException e) {
            log.error("sendToQueue1 Exception ",e);
        }
    }

    @ApiOperation(value = "发送同步消息到topic Q1", httpMethod = "POST")
    @RequestMapping(value = "sendsyncToQueue1")
    public SendResult sendsyncToQueue1(String msg) {
        Message message = MessageBuilder.withPayload(msg).build();
        //发送同步消息（阻塞当前线程，等待broker响应发送结果，这样不太容易丢失消息）
        SendResult sendResult = mqTemplate.syncSend(MQConstant.TOPIC_ONE, message);
        log.info("sendsyncToQueue1 sendResult={}", sendResult);
        return sendResult;
    }

    @ApiOperation(value = "发送异步消息到topic Q1", httpMethod = "POST")
    @RequestMapping(value = "sendAsyncToQueue1")
    public void sendAsyncToQueue1(String msg) {
        Message message = MessageBuilder.withPayload(msg).build();
        //发送同步消息（阻塞当前线程，等待broker响应发送结果，这样不太容易丢失消息）
        mqTemplate.asyncSend(MQConstant.TOPIC_ONE, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("sendAsyncToQueue1 sendResult={}", sendResult);
            }

            @Override
            public void onException(Throwable e) {
                log.error("发送异步消息到topic Q1 异常 ", e);
            }
        });
    }

    @ApiOperation(value = "发送消息到topic Q2 with tag", httpMethod = "POST")
    @RequestMapping(value = "sendToQueue2WithTag")
    public void sendToQueue2WithTag(String msg, String tag) {
        String topic = MQConstant.TOPIC_TWO.concat(":").concat(tag);
        //普通发送 发送带tag的消息，直接在topic后面加上":tag"
        mqTemplate.send(topic, MessageBuilder.withPayload(msg).build());
    }



    @ApiOperation(value = "发送事务消息到topic Q", httpMethod = "POST")
    @RequestMapping(value = "sendToQueue")
    public TransactionSendResult sendTransToQueue(String msg) {
        String uid = UUID.randomUUID().toString();
        Message message = MessageBuilder.withPayload(msg).
                setHeader(RocketMQHeaders.TRANSACTION_ID, uid).
                setHeader("msg_id",UUID.randomUUID().toString()+"TX").
                build();
        TransactionSendResult sendResult = mqTemplate.sendMessageInTransaction(MQConstant.TOPIC_TRANSACTION, message, uid);
        log.info("发送事务消息到topic result = {}",sendResult);
        return sendResult;
    }

}